#ifndef _MANAGER_HPP_
#define _MANAGER_HPP_

#include <atomic>
#include <cstdlib>
#include <cstring>
#include <fcntl.h>
#include <iostream>
#include <list>
#include <mutex>
#include <pthread.h>
#include <stdint.h>
#include <string>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <unistd.h>
#include <vector>
#include <dirent.h>
#include <fstream>
#include <sstream>
#include <signal.h>
#include <XLogger.hpp>
#include <Config.hpp>

#include <MqttCommon.hpp>
#include <MqttClient.hpp>
#include <MqttMsgParse.hpp>

static const std::vector<std::string> localSubTopicInit = {
	"app/+/sys_manager/rack",
	"app/+/sys_manager/wack",
	"app/+/0/info"
};

class Manager {
public:
    Manager(const std::string &rootPath) {
        m_rootPath = rootPath;
        pathInit();
        appInit();
        mqttInit();

		/* 守护线程 */
		std::thread([&] {
		  monitorThread();
		}).detach();

		/* mqtt 线程处理函数 */
		std::thread([&] {
		  parseMqttMsgThread();
		}).detach();
    }

    ~Manager() = default;

private:
    static bool isDirExist(std::string dir) {
        return !access(dir.c_str(), F_OK);
    }

    static bool pathCreate(const std::string& path) {
        std::string currentPath;
        size_t startPos = 0;
        size_t endPos = path.find('/');

        while (endPos != std::string::npos) {
            currentPath += path.substr(startPos, endPos - startPos);
            std::string dir = m_rootPath + currentPath; //添加根目录
            if (!isDirExist(dir)) {
                if (mkdir(dir.c_str(), 0777) != 0) {
                    LOG_ERROR("can't create dir : {}", dir.c_str());
                    return false;
                }
            }
            currentPath += '/';
            startPos = endPos + 1;
            endPos = path.find('/', startPos);
        }

        currentPath += path.substr(startPos);
        std::string dir = m_rootPath + currentPath; //添加根目录
        if (!isDirExist(dir)) {
            if (mkdir(dir.c_str(), 0777) != 0) {
                LOG_ERROR("can't create dir : {}", dir.c_str());
                return false;
            }
        }
        return true;
    }

    static void pathInit(void) {
        LOG_DEBUG("sys_manager version : {}", "v0.0.3");
        pathCreate("sys/sys_agent/profile/modbus");
        pathCreate("sys/sys_manager/file");
        pathCreate("dev/dev_modbus");
        pathCreate("mid/mid_sql");
        pathCreate("mid/ui");
    }

    static bool appInit(void) {
        config_.reset(new AppConfig("/home/chint/sys/sys_manager/file/app_launch.yaml"));
        config_->print();
        return true;
    }

/** --------------------------------MQTT---------------------------------------- */

	/* 本地连接回调函数 */
	static void localOnConnect(struct mosquitto*, void*, int rc) {
		if (rc == 0) {
			/* 订阅主题 */
			for (const auto& topic: localSubTopicInit) {
				localClient_->subscribe(topic);
			}
			localClientIsConnected_ = true;
			LOG_DEBUG("Connected to Local Mqtt broker.");
		}
		else {
			localClientIsConnected_ = false;
			LOG_DEBUG("Connection failed: {}", mosquitto_strerror(rc));
		}
	}

	/* 本地断开连接回调函数 */
	static void localOnDisconnect(struct mosquitto*, void*, int rc) {
		localClientIsConnected_ = false;
		LOG_DEBUG("Disconnected from Local Mqtt broker.");
	}

	/* 本地接收到消息后的回调函数 */
	static void localMsgCallback(struct mosquitto* mosq, void* obj, const struct mosquitto_message* msg) {
		updateMqttReceiveQueue({ msg->topic },
			{ static_cast<const char*>(msg->payload), static_cast<size_t>(msg->payloadlen) });
	}

	/* 更新 Mqtt 接收队列 */
	static void updateMqttReceiveQueue(const std::string& topic, const std::string& payload) {
		if (receiveMqttMsgQueue_.size() > kMqttQueueLength_) {
			receiveMqttMsgQueue_.pop();
		}
		receiveMqttMsgQueue_.emplace(topic, payload);
		std::lock_guard<std::mutex> _lck(receiveMqttMutex_);
		isRead_ = true;
		receiveMqttCV_.notify_one();
	}

	/* Mqtt 函数 */
	static void parseMqttMsgThread() {
		while (run_) {
			std::unique_lock<std::mutex> lock(receiveMqttMutex_);
			receiveMqttCV_.wait(lock, [] { return isRead_; });
			isRead_ = false;

			while (!receiveMqttMsgQueue_.empty()) {
				const auto& msg = receiveMqttMsgQueue_.front();
				mqttMsgParse_->parseMsg(msg);
				receiveMqttMsgQueue_.pop();
			}

			// 添加延时操作，让出CPU执行时间给其他线程
        	std::this_thread::sleep_for(std::chrono::milliseconds(10));
		}
	}

    static bool mqttInit(void) {
        MqttConnectInfo localInfo = {
            .server = "127.0.0.1",
            .port = 1883,
            .clientID = "sys_manager",
            .user = "",
            .passwd = "",
            .cleanSession = true,
            .keepAlive = 60
        };

		localClient_.reset(new MqttClient(
			localInfo.clientID,
			localInfo.server,
			localInfo.port,
			localInfo.user.c_str(),
			localInfo.passwd.c_str(),
			localInfo.cleanSession,
			localOnConnect,
			localOnDisconnect,
			localMsgCallback));

		/* 连接服务器 */
		localClient_->connect(localInfo.keepAlive);

        // 实例化MqttMsgParse类，并将已实例化的AppConfig类的引用传递给MqttMsgParse类的构造函数
        mqttMsgParse_ = std::make_shared<MqttMsgParse>(config_);

        return true;
    }

    static bool killProcessByName(const std::string& programName) {
        std::string command = "pkill " + programName;
        int result = system(command.c_str());
        if (result == -1) {
            // 执行命令失败
            std::cerr << "Failed to execute the command." << std::endl;
            return false;
        } else if (WIFEXITED(result)) {
            int exitCode = WEXITSTATUS(result);
            if (exitCode == 0) {
                // 命令执行成功
                std::cout << "Process " << programName << " terminated successfully." << std::endl;
                return true;
            } else {
                // 命令执行失败
                std::cerr << "Failed to terminate process " << programName << " with exit code: " << exitCode << std::endl;
                return false;
            }
        } else {
            // 命令异常终止
            std::cerr << "Command terminated abnormally." << std::endl;
            return false;
        }
    }

    // 检查进程是否已经ping
    static bool pingProcess(int& programStatus, int& programPingCnt, const std::string& programName) {
        if (programStatus == 0) {
            msgAppPingRead(programName);
            if (++programPingCnt > 3) { //OT @15s
                programPingCnt = 0;
                LOG_ERROR("Process : {} mqtt heartbeat err", programName);
                killProcessByName(programName);
                return false;
            }
        } else if (programStatus == 1) {
            LOG_DEBUG("Process : {} mqtt heartbeat ok", programName);
            programStatus = 0;
            programPingCnt = 0;
        }
        return true;
    }

    // 检查进程是否已经存在
    static bool isProcessRunning(const std::string& programPath) {
        std::string command = "ps aux | grep -e " + programPath + " | grep -v grep";
        LOG_DEBUG("isProcessRunning command : {}", command);
        FILE* pipe = popen(command.c_str(), "r");
        if (!pipe) {
            LOG_ERROR("Failed to execute the command.");
            return false;
        }
        char buffer[128];
        std::string result;
        while (fgets(buffer, sizeof(buffer), pipe) != nullptr) {
            result += buffer;
        }
        pclose(pipe);
        LOG_DEBUG("isProcessRunning command : {}", result);
        return !result.empty();
    }

    // 启动目标进程
    static bool startProcess(const std::string& programPath, const std::string& programArgs) {
        std::string command = programPath + " " + programArgs + " " + "&";
        LOG_DEBUG("command : {}", command);
        int result = system(command.c_str());
        if (result == -1) {
            // 执行命令失败
            std::cerr << "Failed to execute the command." << std::endl;
            LOG_ERROR("Failed to execute the command.");
            return false;
        } else if (WIFEXITED(result)) {
            int exitCode = WEXITSTATUS(result);
            if (exitCode == 0) {
                // 命令执行成功
                LOG_DEBUG("Process {}  terminated successfully.", command);
                //std::cout << "Process " << command << " terminated successfully." << std::endl;
                return true;
            } else {
                // 命令执行失败
                LOG_ERROR("Failed to terminate process {} with exit code: ", command, exitCode);
                //std::cerr << "Failed to terminate process " << command << " with exit code: " << exitCode << std::endl;
                return false;
            }
        } else {
            // 命令异常终止
            LOG_ERROR("Command terminated abnormally.");
            //std::cerr << "Command terminated abnormally." << std::endl;
            return false;
        }
        return true;
    }

    static void monitorThread() {
        while (true) {
            for (auto& app : config_->appList) {
                // 检查进程是否已经存在
                if (isProcessRunning(app.file)) {
                    LOG_DEBUG("Process : {} is runing...", app.name);
                    pingProcess(app.status, app.pingCnt, app.name);
                } else {
                    LOG_ERROR("Process : {} is restarting...", app.name);
                    // 启动目标进程
                    startProcess(app.file, app.cmd);
                }
            }

            // 每隔一段时间检查一次目标进程状态
            std::this_thread::sleep_for(std::chrono::seconds(5));
        }
    }

    //读应用ping消息
    static bool msgAppPingRead(const std::string& programName) {
        struct timeval tv;
        nlohmann::json pingJson;
        pingJson["tick"] = 1;
        pingJson["time"] = getLocalTimeStampMs();
        pingJson["src"] = "sys_manager";
		nlohmann::json& pointsArray = pingJson["app"]["points"];
        nlohmann::json point1;
        point1["key"] = "log";
        pointsArray.push_back(point1);
        nlohmann::json point2;
        point2["key"] = "version";
        pointsArray.push_back(point2);
        nlohmann::json point3;
        point3["key"] = "date";
        pointsArray.push_back(point3);
        std::string pingTopic = "app/sys_manager/" + programName + "/read";
		localClient_->publish(pingTopic, pingJson.dump());
		return true;
    }

private:
    static std::shared_ptr<AppConfig> config_;
    static std::string m_rootPath;
    static std::shared_ptr<MqttClient> localClient_;
    static bool localClientIsConnected_;

	static std::queue<Msg> receiveMqttMsgQueue_;                /* mqtt receive msg queue */
	static std::mutex receiveMqttMutex_;                        /* mqtt receive mutex */
	static std::condition_variable receiveMqttCV_;              /* mqtt receive condition variable */
	static bool isRead_;                                        /* prevent spurious wakeup */
	static const int kMqttQueueLength_ = 1024;                  /* mqtt receive queue max length */
	static bool run_;

    static std::shared_ptr<MqttMsgParse> mqttMsgParse_; // 声明一个MqttMsgParse类的实例指针
};

//使用类名限定符来引用，明确指定访问的是类的静态成员变量，而不是全局变量
std::shared_ptr<AppConfig> Manager::config_;
std::string Manager::m_rootPath;
std::shared_ptr<MqttClient> Manager::localClient_;
bool Manager::localClientIsConnected_{ false};

std::queue<Msg> Manager::receiveMqttMsgQueue_;
std::mutex Manager::receiveMqttMutex_;
std::condition_variable Manager::receiveMqttCV_;
bool Manager::isRead_ = false;
bool Manager::run_{ true };

std::shared_ptr<MqttMsgParse> Manager::mqttMsgParse_;

#endif
